// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/common/base/exceptions.h>
#include <pollux/expression/vector_readers.h>
#include <pollux/expression/vector_writers.h>

namespace kumo::pollux::exec {
    namespace {
        // TODO: support fast path for unknown.
        bool fastPathSupported(const TypePtr &type) {
            return type->isPrimitiveType() && type->kind() != TypeKind::UNKNOWN;
        }

        template<TypeKind kind>
        bool constexpr fastPathSupportedStatic() {
            return TypeTraits<kind>::isPrimitiveType && kind != TypeKind::UNKNOWN;
        }

        // Base case for primitives.
        template<TypeKind T>
        void copy_from_internal(GenericWriter &out, const GenericView &in) {
            // Maybe we should print a warning asking the user to have a fast path
            // specialization for primitives and not to use this?.
            using simple_element_t = typename KindToSimpleType<T>::type;
            out.castTo<simple_element_t>() = in.castTo<simple_element_t>();
        }

        template<>
        void copy_from_internal<TypeKind::VARBINARY>(
            GenericWriter &out,
            const GenericView &in) {
            out.castTo<Varbinary>().copy_from(in.castTo<Varbinary>());
        }

        template<>
        void copy_from_internal<TypeKind::VARCHAR>(
            GenericWriter &out,
            const GenericView &in) {
            out.castTo<Varchar>().copy_from(in.castTo<Varchar>());
        }

        // Fast path when array elements are primitives.
        template<TypeKind T>
        void copy_from_internal_array_fast(GenericWriter &out, const GenericView &in) {
            if constexpr (fastPathSupportedStatic<T>()) {
                using simple_element_t = typename KindToSimpleType<T>::type;
                out.castTo<Array<simple_element_t> >().copy_from(
                    in.castTo<Array<simple_element_t> >());
            } else {
                POLLUX_UNREACHABLE("unsupported type dispatched");
            }
        }

        template<>
        void copy_from_internal<TypeKind::ARRAY>(
            GenericWriter &out,
            const GenericView &view) {
            //   Fast path for when the array element is primitive.
            if (fastPathSupported(out.type()->childAt(0))) {
                TypeKind kind = out.type()->childAt(0)->kind();
                POLLUX_DYNAMIC_TYPE_DISPATCH_ALL(
                    copy_from_internal_array_fast, kind, out, view);
            } else {
                auto &writer = out.castTo<Array<Any> >();
                auto arrayView = view.castTo<Array<Any> >();
                writer.copy_from(arrayView);
            }
        }

        template<TypeKind KeyT, TypeKind ValueT>
        void copy_from_internal_map_fast_keys_values(
            GenericWriter &out,
            const GenericView &in) {
            if constexpr (
                fastPathSupportedStatic<KeyT>() && fastPathSupportedStatic<ValueT>()) {
                using simple_key_t = typename KindToSimpleType<KeyT>::type;
                using simple_value_t = typename KindToSimpleType<ValueT>::type;

                auto &writer = out.castTo<Map<simple_key_t, simple_value_t> >();
                auto mapView = in.castTo<Map<simple_key_t, simple_value_t> >();
                writer.copy_from(mapView);
            } else {
                POLLUX_UNREACHABLE("unsupported type dispatched");
            }
        }

        template<TypeKind KeyT>
        void copy_from_internal_map_fast_keys(
            GenericWriter &out,
            const GenericView &in) {
            if (fastPathSupported(out.type()->childAt(1))) {
                TypeKind kind = out.type()->childAt(1)->kind();
                POLLUX_DYNAMIC_SCALAR_TEMPLATE_TYPE_DISPATCH(
                    copy_from_internal_map_fast_keys_values, KeyT, kind, out, in);
            } else {
                if constexpr (fastPathSupportedStatic<KeyT>()) {
                    using simple_key_t = typename KindToSimpleType<KeyT>::type;

                    auto &writer = out.castTo<Map<simple_key_t, Any> >();
                    auto mapView = in.castTo<Map<simple_key_t, Any> >();
                    writer.copy_from(mapView);
                } else {
                    POLLUX_UNREACHABLE("unsupported type dispatched");
                }
            }
        }

        template<TypeKind ValueT>
        void copy_from_internal_map_fast_values(
            GenericWriter &out,
            const GenericView &in) {
            if constexpr (fastPathSupportedStatic<ValueT>()) {
                using simple_value_t = typename KindToSimpleType<ValueT>::type;
                auto &writer = out.castTo<Map<Any, simple_value_t> >();
                auto mapView = in.castTo<Map<Any, simple_value_t> >();
                writer.copy_from(mapView);
            } else {
                POLLUX_UNREACHABLE("unsupported type dispatched");
            }
        }

        template<>
        void copy_from_internal<TypeKind::MAP>(
            GenericWriter &out,
            const GenericView &view) {
            if (fastPathSupported(out.type()->childAt(0))) {
                TypeKind kind = out.type()->childAt(0)->kind();
                POLLUX_DYNAMIC_TYPE_DISPATCH_ALL(
                    copy_from_internal_map_fast_keys, kind, out, view);
            } else if (fastPathSupported(out.type()->childAt(1))) {
                TypeKind kind = out.type()->childAt(1)->kind();
                POLLUX_DYNAMIC_TYPE_DISPATCH_ALL(
                    copy_from_internal_map_fast_values, kind, out, view);
            } else {
                auto &writer = out.castTo<Map<Any, Any> >();
                auto mapView = view.castTo<Map<Any, Any> >();
                writer.copy_from(mapView);
            }
        }

        template<>
        void copy_from_internal<TypeKind::ROW>(
            GenericWriter &out,
            const GenericView &in) {
            auto dyanmicRowView = in.castTo<DynamicRow>();
            auto &dynamicRowWriter = out.castTo<DynamicRow>();
            for (int i = 0; i < dyanmicRowView.size(); i++) {
                auto fieldGenericView = dyanmicRowView.at(i);

                if (fieldGenericView.has_value()) {
                    TypeKind kind = fieldGenericView->kind();
                    POLLUX_DYNAMIC_TYPE_DISPATCH(
                        copy_from_internal,
                        kind,
                        dynamicRowWriter.get_writer_at(i),
                        fieldGenericView.value());
                } else {
                    dynamicRowWriter.set_null_at(i);
                }
            }
        }
    } // namespace

    void GenericWriter::copy_from(const GenericView &view) {
        TypeKind kind = this->kind();
        POLLUX_DYNAMIC_TYPE_DISPATCH_ALL(copy_from_internal, kind, *this, view);
    }
} // namespace kumo::pollux::exec
